// Copyright (c) 2011 The Kiwii Authors. All rights reserved.
// Use of this source code is governed by the New BSD License that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "comm/Kiwii.hxx"
#include "net/TcpConnection.hxx"
#include "net/Networking.hxx"
#include "comm/Exception.hxx"
#include <boost/assert.hpp>

namespace kiwii {
  namespace net {

    TcpConnection::TcpConnection(int connfd, EventLoopBase& el) : 
      connfd_(connfd), 
      el_(el), 
      inputBuffer_(new char[INIT_BUFFER_SIZE]), inputBufferSize_(INIT_BUFFER_SIZE), inTo_(0),
      outputBuffer_(new char[INIT_BUFFER_SIZE]), outputBufferSize_(INIT_BUFFER_SIZE), outTo_(0) {

      setNonBlocking(connfd_);

      el_.setIOEvent(connfd_, EV_READABLE, [this](EventLoopBase& el, int fd, unsigned int firedEvents) {
	  kassert(fd == connfd_);
	  KIWII_UNUSED(el);
	  KIWII_UNUSED(firedEvents);

	  //Here we continue reading input until there is no more data left 
	  //in the socket receive buffer, to accommodate both of the two kind of
	  //io multiplexing mode: edge-triggered one and level-triggered one.
	  for (;;) {
	    auto n = read(fd, &inputBuffer_[inTo_], inputBufferSize_ - inTo_);
	    if (n < 0) {
	      if (errno == EAGAIN) 
		break;
	      else
		throw comm::Exception("read() failed:") << boost::errinfo_errno(errno);
	    } else if (n == 0) {
	      inTo_ = 0;
	      break;
	    } else {
	      inTo_ += n;
	      if (0 == (inputBufferSize_ - inTo_)) {
		//There is more data to be read in, so expand the input buffer.
		inputBufferSize_ *= 2;
		std::unique_ptr<char[]> temp(new char[inputBufferSize_]);
		memcpy(&temp[0], &inputBuffer_[0], inTo_);
		inputBuffer_.swap(temp);
	      } else {
		//0 < available buffer size means there is no more data left.
		break;
	      }
	    }
	  }

	  //Call message handler
	  auto consume = msgHandler_(&inputBuffer_[0], inTo_);
	  if (consume <= inTo_) {
	    //move the rest data to buffer head.
	    inTo_ -= consume;
	    if (inTo_ != 0) {
	      memmove(&inputBuffer_[0], &inputBuffer_[consume], inTo_);
	    }
	  } else {
	    throw comm::Exception("message handler consumed excessive buffer");
	  }
	});

      el_.setIOEvent(connfd_, EV_WRITABLE, [this](EventLoopBase&, int fd, unsigned int) {
	  kassert(fd == connfd_);
	  if (outTo_ != 0) {
	    auto n = write(connfd_, &outputBuffer_[0], outTo_);
	    if (n < 0 && errno != EAGAIN) {
	      throw comm::Exception("write() failed:") << boost::errinfo_errno(errno);
	    } else {
	      outTo_ -= n;
	      if (outTo_ != 0) {
		memmove(&outputBuffer_[0], &outputBuffer_[n], outTo_);
	      }
	    }
	  }
	});
    }

    void TcpConnection::send(char const* buf, std::size_t size) {
      if (size > outputBufferSize_ - outTo_) {
	while (size > outputBufferSize_ - outTo_) {
	  outputBufferSize_ *= 2;
	}
	std::unique_ptr<char[]>temp(new char[outputBufferSize_]);
	memcpy(&temp[0], &outputBuffer_[0], outTo_);
	outputBuffer_.swap(temp);
      }
      memcpy(&outputBuffer_[outTo_], buf, size);
      outTo_ += size;
    }

    TcpConnection::~TcpConnection() {
      el_.delIOEvent(connfd_, EV_ALL);
      close(connfd_);
    }

  }
}
